package com.bgds.sys.influx.utils;

import com.bgds.sys.common.constant.BizConstant;
import com.bgds.sys.config.influx.InfluxConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @Description TODO
 * @Author lj
 * @create: 2020-07-23 11:26
 */
@Slf4j
@Component
public class InfluxUtils {
    @Resource
    private InfluxDB influxDb;
    @Resource
    private InfluxConfig influxConfig;

    public QueryResult querySql(String sql) {
        Query query = new Query(sql, influxConfig.getInfluxDatabase());
        QueryResult result = influxDb.query(query);
        log.info(result.toString());
        return result;
    }

    /**
     * 创建influx数据库
     */
    public void createInfluxDatabase(String databaseName) {
        Query query = new Query("CREATE DATABASE " + databaseName, databaseName);
        influxDb.query(query);
    }

    /**
     * 创建保存策略
     * @param databaseName
     * @param policyName
     */
    public void createPolicy(String databaseName,String policyName) {
        Query query = new Query("CREATE RETENTION POLICY "+policyName+" ON "+databaseName+" DURATION 24h REPLICATION 1 ", databaseName);
        influxDb.query(query);
    }
    /**
     * 修改保存策略
     * @param databaseName
     * @param policyName
     */
    public void alterPolicy(String databaseName,String policyName) {
        Query query = new Query("ALTER RETENTION POLICY "+policyName+" ON "+databaseName+" DURATION 24h DEFAULT ", databaseName);
        influxDb.query(query);
    }
    /**
     * 删除保存策略
     * @param databaseName
     * @param policyName
     */
    public void deletePolicy(String databaseName,String policyName) {
        Query query = new Query("DROP RETENTION POLICY "+policyName+" ON "+databaseName, databaseName);
        influxDb.query(query);
    }
    /**
     * 插入数据:
     * tag的使用
     */
    public void writeRecord(String measurement, Map<String, String> tags, Map<String, Object> fields) {
        influxDb.write(Point.measurement(measurement)
                .tag(tags)
                .fields(fields)
                .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
                .build());
    }

    /**
     * 查询
     *
     * @param command 查询语句
     */
    public QueryResult query(String command) {
        return influxDb.query(new Query(command, influxConfig.getInfluxDatabase()));
    }

    /**
     * 解析后返回map的查询
     */
    public List<Map<String, Object>> queryForMap(String command) {
        List<Map<String, Object>> list = new ArrayList<>();
        QueryResult queryResult = influxDb.query(new Query(command, influxConfig.getInfluxDatabase()));

        if (null != queryResult && queryResult.getResults().get(0).getSeries()!=null) {

            for (QueryResult.Result result : queryResult.getResults()) {

                List<QueryResult.Series> series = result.getSeries();

                for (QueryResult.Series s : series) {
                    List<List<Object>> values = s.getValues();
                    List<String> columns = s.getColumns();
                    list = getQueryData(columns, values);
                }
            }
        } else {
//            log.error("influxDB未查询到相关数据:" + command);
            return list;

        }
        return list;
    }

    /**
     * 将influxdb的结果解析成键值对list
     *
     * @param colList valuesList
     */
    private List<Map<String, Object>> getQueryData(List<String> colList,
                                                   List<List<Object>> valuesList) {
        List<Map<String, Object>> list = new ArrayList<>();

        for (int i = 0; i < valuesList.size(); i++) {
            Map<String, Object> map = new HashMap<>(10);
            List<Object> oList = valuesList.get(i);

            for (int j = 0; j < colList.size(); j++) {
                map.put(colList.get(j), oList.get(j) == null ? "" : oList.get(j));
            }
            list.add(map);
        }

        return list;
    }

    /**
     * 插入
     *
     * @param measurement 表
     * @param tags        标签
     * @param fields      字段
     */
    public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields) {
        Point.Builder builder = Point.measurement(measurement);
        builder.time(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
       // builder.tag(tags);
        builder.fields(fields);
        influxDb.write(influxConfig.getInfluxDatabase(), "", builder.build());
    }

    /**
     * @param fields 字段
     */
    public void insert(Map<String, Object> fields) {
        Point.Builder builder =
                Point.measurement(fields.get(BizConstant.DEVICE_NAME).toString());
        builder.fields(fields);
        influxDb.write(influxConfig.getInfluxDatabase(), "", builder.build());
    }

    /**
     * 删除
     *
     * @param command 删除语句
     * @return 返回错误信息
     */
    public String deleteMeasurementData(String command) {
        QueryResult result = influxDb.query(new Query(command, influxConfig.getInfluxDatabase()));
        return result.getError();
    }

    /**
     * 批量插入
     *
     * @param measurement 表
     * @param tags        标签
     * @param fieldsList  字段
     */
    public void batchInsert(String measurement, Map<String, String> tags, List<Map<String, Object>> fieldsList) {
        BatchPoints.Builder batchBuilder = BatchPoints.database(influxConfig.getInfluxDatabase());

        for (Map<String, Object> map : fieldsList) {
            Point.Builder builder = Point.measurement(measurement);
            tags.put("instanceId", map.get("instanceId").toString());
            builder.time((long) map.get("currentTime"), TimeUnit.NANOSECONDS);
            builder.tag(tags);
            builder.fields(map);
            batchBuilder.point(builder.build());
        }

        log.debug("influxDB批量插入：" + batchBuilder.build().toString());
        influxDb.write(batchBuilder.build());
    }

    /**
     * 批量插入influxDb数据
     * @param fieldsList
     */
    public void batchInsert(List<Map<String, Object>> fieldsList) {

        fieldsList.stream()
                .filter(e -> StringUtils.isNotBlank(e.get(BizConstant.MODEL_NAME).toString()))
                .sorted(this::complexComparator)
                .forEach(map -> {
                    Point.Builder builder = Point
                            .measurement(map.get(BizConstant.MODEL_NAME).toString());
                    map.remove(BizConstant.MODEL_NAME);
                    builder.time((Long) (map.get(BizConstant.TIME)),
                            TimeUnit.MILLISECONDS);
                    builder.tag(BizConstant.DEVICE_CODE, (String) map.get(BizConstant.DEVICE_CODE));
                    map.remove(BizConstant.DEVICE_CODE);
                    builder.fields(map);
                    log.debug("刷库之前influx{}", builder.build());
                    influxDb.write(influxConfig.getInfluxDatabase(), "", builder.build());
                });
    }

    private int complexComparator(Map<String, Object> map1,
                                  Map<String, Object> map2) {
        int c = getMesurement(map1).compareTo(getMesurement(map2));
        if (c != 0) {
            return c;
        }

        return getTime(map1).compareTo(getTime(map2));
    }

    private String getMesurement(Map<String, Object> map) {
        return (String) map.get(BizConstant.MODEL_NAME);
    }

    private Long getTime(Map<String, Object> map) {
        return (Long) (map.get(BizConstant.TIME));
    }


}
